package com.innodealing.kafka.listener;

import com.innodealing.kafka.model.KafkaMessage;
import com.innodealing.onshore.datasync.plus.constant.KafkaConstants;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.listener.BatchAcknowledgingMessageListener;
import org.springframework.kafka.support.Acknowledgment;

import java.util.ArrayList;
import java.util.List;

/**
 * Kafka 批量消息监听器
 *
 * @author Yinpenghai
 */
public class KafkaBatchMessageListener implements BatchAcknowledgingMessageListener<Object, Object> {

    private final HermesBatchOrderlyListener listener;

    /**
     * 构造函数
     *
     * @param listener 监听器
     */
    public KafkaBatchMessageListener(HermesBatchOrderlyListener listener) {
        this.listener = listener;
    }

    @Override
    public void onMessage(List<ConsumerRecord<Object, Object>> data, Acknowledgment acknowledgment) {
        List<KafkaMessage> messages = new ArrayList<>();

        for (ConsumerRecord<Object, Object> record : data) {
            // 跳过同步平台特定的消息
            if (!isRecordValid(record)) {
                continue;
            }

            // 转换 ConsumerRecord 为 KafkaMessage
            KafkaMessage kafkaMessage = new KafkaMessage(
                    record.topic(),
                    record.partition(),
                    record.offset(),
                    record.timestamp(),
                    record.key() != null ? record.key().toString() : null,
                    record.value() != null ? record.value().toString() : null
            );
            messages.add(kafkaMessage);
        }

        if(!messages.isEmpty()){
            // 调用使用者定义的 onMessage 处理消息
            listener.onMessage(messages);
        }

        // 手动ACK
        acknowledgment.acknowledge();
    }

    private boolean isRecordValid(ConsumerRecord<?, ?> record) {
        Object value = record.value();
        return value != null && !KafkaConstants.KAFKA_CREATE_TOPIC_MESSAGE.equals(value.toString());
    }
}
